package com.xian.scala.stream

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer

import java.util.Properties


/**
 * flink 消费kafka信息
 */
object GetKafkaMes {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val props = new Properties
    props.setProperty("bootstrap.servers","172.18.9.1:9092,172.18.9.2:9092,172.18.9.3:9092")
    val lines: DataStream[String] = env.addSource(new FlinkKafkaConsumer[String]("jemson-topic", new SimpleStringSchema(), props))

    lines.print()


    env.execute()

  }
}
